package udpInvoker.main.lib;

import udpInvoker.main.*;
import udpInvoker.util.UdpUtils;
import udpInvoker.exceptions.NoInvokableResponseException;

import java.io.IOException;
import java.util.Date;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * This class should be used for invoking remote methods.  
 * Copyright 2008  Espen Skjervold, FFI
 */
public class Invoker implements MessageSubscriber {


    private String password;
    private UdpMulticastReceiver udpMulticastReceiver;
    private BlockingQueue messageQueue;
    private int pingTimeout = 1000;
    private boolean pingBeforeInvocation =true;
    private int transportType=1;
    private String unicastAddress;
    private int port;



    public static final int MULTICAST=1;
    public static final int UNICAST=2;



    public Invoker() {

        messageQueue = new LinkedBlockingQueue();
        udpMulticastReceiver = UdpMulticastReceiver.getInstance();
        udpMulticastReceiver.subscribe(this);

    }

//    lib void setUnicastAddress(String address, int port) {
//        unicastAddress=address;
//        this.port=port;
//
//        if (address!=null) {
//            transportType=UNICAST;
//            UdpUnicastReceiver udpUnicastReceiver = UdpUnicastReceiver.getInstance(unicastAddress, port);
//            udpUnicastReceiver.subscribe(this);
//        }
//        else {
//            transportType=MULTICAST;
//            udpMulticastReceiver.setUnicastAddress(null, 0);
//        }
//    }

     /**
     * Sets the ping timout interval, when isOnline() is invoked.
     * @param millis The timeout interval in milliseconds
     */
    public void setPingTimeoutInMillis(int millis) {
        pingTimeout = millis;
    }


     /**
     * Invokes a public method in a class extending Invokable.
     * @param methodName The name of the method to be invoked. Can be a fully qyualifying name (e.g. "com.main.MyClass.myMethod" or the simple name (e.g. "myMethod")
     * @param parameters An arbitrary number of arguments of any type extending Object. E.g String, Integer, ArrayList, CustomClass etc.
     * @return  An instance of Object, should be casted to an object of the type the method actually returns. Will be null when method returns void.
     * @throws Exception The invocation was not successful (no matching Invokable found etc. )
     */
    public Object invoke(String methodName, Object... parameters) throws Exception {
        return doInvoke(methodName, false,  parameters);
     }


     /**
     * Invokes a remote, public method in a class extending Invokable. The invocation- and response-message is encrypted.
     * @param methodName The name of the method to be invoked. Can be a fully qyualifying name (e.g. "com.main.MyClass.myMethod" or the simple name (e.g. "myMethod")
      *@param password A string representing the password used for the encryption. The same password should by use by setPassword() in the Invokable's constructor.
      * @param parameters An arbitrary number of arguments of any type extending Object. E.g String, Integer, ArrayList, CustomClass etc.
     * @return  An instance of Object, should be casted to an object of the type the method actually returns. Will be null when method returns void.
     * @throws Exception The invocation was not successful (no matching Invokable found etc. )
     */
    public Object invokeSecure(String methodName, String password,  Object... parameters) throws Exception {
        this.password = password;
        return doInvoke(methodName, true,  parameters);
     }

     /**
     * Checks the remote method availability by sending a ping-message and reveiving a response message.
     * @param methodName The name of the method to check. Can be a fully qyualifying name (e.g. "com.main.MyClass.myMethod" or the simple name (e.g. "myMethod")
     * @param parameters An arbitrary number of arguments of any type extending Object. These must be present to correctly match the method specified, but may be "empty" objects (e.g "new String()");
     * @return  A boolean value indicating whether the method is available.    
     */
    public boolean isOnline(String methodName, Object... parameters) {
        return invokableIsResponding(methodName, false, parameters);
    }


     /**
     * Checks the remote method availability by sending a ping-message and reveiving a response message. Uses encrypted communication.
     * @param methodName The name of the method to check. Can be a fully qyualifying name (e.g. "com.main.MyClass.myMethod" or the simple name (e.g. "myMethod")
     * @param parameters An arbitrary number of arguments of any type extending Object. These must be present to correctly match the method specified, but may be "empty" objects (e.g "new String()");
     * @return  A boolean value indicating whether the method is available.
     */
     public boolean isOnlineSecure(String methodName, String password, Object... parameters) {
        this.password = password;
        return invokableIsResponding(methodName, true, parameters);
    }



    private Object doInvoke(String methodName, boolean useEncryption, Object... parameters) throws Exception {


        if(pingBeforeInvocation && !invokableIsResponding(methodName, useEncryption, parameters))
            throw new NoInvokableResponseException();


        InvocationMessage invocationMessage = UdpUtils.createInvocationMessage(methodName, parameters);


        byte[] buffer = Base64.encodeObject(invocationMessage, Base64.GZIP).getBytes();

        if (useEncryption) {
            buffer = UdpUtils.encryptBuffer(buffer, password);
        }


        UdpUtils.sendBuffer(buffer, transportType, unicastAddress, port);

        
        Object response = receiveResponse(invocationMessage);


        if (isException(response)) {
            Exception e = (Exception) response;
            throw e;
        }
        else
            return response;

    }


    private boolean invokableIsResponding(String methodName, boolean useEncryption, Object[] parameters){

        int retries = 3;

        for (int i=0;i<retries;i++) {
            //System.out.println("try: " + (i+1)  + "/"+ retries);
            if (doInvokableIsResponding(methodName, useEncryption, parameters))
                return true;
        }


        return false;
    }

    private boolean doInvokableIsResponding(String methodName, boolean useEncryption, Object[] parameters){

        boolean responseOk=false;

        InvocationMessage pingMessage = UdpUtils.createPingMessage(methodName, parameters);

        try {
            byte[] buffer = Base64.encodeObject(pingMessage, Base64.GZIP).getBytes();

            if (useEncryption) {
                buffer = UdpUtils.encryptBuffer(buffer, password);
            }

            UdpUtils.sendBuffer(buffer, transportType, unicastAddress, port);


            long startTime = new Date().getTime();

            while (new Date().getTime()< startTime + pingTimeout)  {                
                byte[] returnBuffer = (byte[]) messageQueue.take();
                InvocationMessage message = UdpUtils.deSerializeInvocationMessage(returnBuffer, password);
                if (message.getType()== InvocationMessage.PONG && message.getSerialNumber()==pingMessage.getSerialNumber()) {
                    //System.out.println("Pong-message received");
                    responseOk=true;
                    break;
                }
                else {
                    messageQueue.put(returnBuffer); //not a pong message, put back on queue
                    //System.out.println("Message reveived: " + message.getType());
                }


                Thread.yield();
            }
        } catch (Exception e) {
            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
        }

        return responseOk;


    }

    private Object receiveResponse(InvocationMessage invocationMessage) throws IOException {
        //receive response
        InvocationMessage responseMessage=null;
        while (true)  {

            try {
                byte[] buffer = (byte[]) messageQueue.take();
                responseMessage= UdpUtils.deSerializeInvocationMessage(buffer, password); //UdpUtils.receiveMessage(multicastSocket, password);
                if (responseMessage.getType()== InvocationMessage.RESPONSE && responseMessage.getSerialNumber()==invocationMessage.getSerialNumber())
                    break;
                else {
                    messageQueue.put(buffer); //put message back on que
                    sleep(20);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            }
        }

        Object response = responseMessage.getResponse();
        return response;
    }

    private ServiceDescription[] receiveServiceResponse(InvocationMessage serviceRequestMessage) throws IOException {
        //receive service response
        InvocationMessage serviceResponseMessage=null;
        ArrayList discoveredServices = new ArrayList();
        long initialTimeout = new Date().getTime() + 4000;
        long lastReceivedTimestamp=-1;
        int noMessagesReceived=0;

        while (new Date().getTime()<= initialTimeout || (lastReceivedTimestamp!=-1 && new Date().getTime()<=(lastReceivedTimestamp + 1000)) )  {

            try {

                 byte[] buffer;


                synchronized (messageQueue) {
                if (messageQueue.peek()!=null) {
                    //System.out.println("take...");
                    buffer = (byte[]) messageQueue.take();
                    //System.out.println("taken.");

                    serviceResponseMessage= UdpUtils.deSerializeInvocationMessage(buffer, password); //UdpUtils.receiveMessage(multicastSocket, password);

                    if (serviceResponseMessage.getType()== InvocationMessage.SERVICE_RESPONSE && serviceResponseMessage.getSerialNumber()==serviceRequestMessage.getSerialNumber())     {
                        ArrayList serviceList = (ArrayList) serviceResponseMessage.getResponse();
                        discoveredServices.addAll(serviceList);
                        lastReceivedTimestamp = new Date().getTime();
                        noMessagesReceived++;

                    }
                    else {
                        messageQueue.put(buffer); //put message back on que
                        sleep(20);
                    }


                }
                else {
                    sleep(100);
                }
                }

                
            } catch (InterruptedException e) {
                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            }
        }

        System.out.println("Discovered messages from " + noMessagesReceived + " JVMs");

        discoveredServices = removeDuplicateServices(discoveredServices);

        ServiceDescription[] services = new ServiceDescription[discoveredServices.size()];
        for (int i=0;i<services.length;i++) {
            services[i]=(ServiceDescription)discoveredServices.get(i);
        }

        return services;
    }

    private ArrayList removeDuplicateServices(ArrayList allServices) {
        HashMap tempHash = new HashMap();

        ArrayList  services = new ArrayList();

        for (int i=0;i<allServices.size();i++) {
            ServiceDescription service = (ServiceDescription) allServices.get(i);
            if (!tempHash.containsKey(service.toString()))  {
                tempHash.put(service.toString(), null);
                services.add(service);
            }
        }

        return services;
    }

    private void sleep(int millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
        }


    }


    private boolean isException(Object response) {

        if (response==null)
            return false; //not an exception

        try {
            Exception e = (Exception) response;
        } catch (Exception e1) {
            return false;
        }

        return true;

    }

     /**
     * Internal method, do not use. 
     */
    public  void addMessageToQue(byte[] buffer) {
        try {
            messageQueue.put(buffer);
        } catch (InterruptedException e) {
            e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
        }
    }


     /**
     * Set to True by default. Forces an availability-check before invoking a remote method. Can be set to false.
     * @param pingBeforeInvocation Boolean value, true or false.    
     */
    public void setPingBeforeInvocation(boolean pingBeforeInvocation) {
        this.pingBeforeInvocation = pingBeforeInvocation;
    }


     /**
     * Subscribes to all messages (Objects) published to the specified topic.
     * @param topicSubscriber Any instance of a class implementing TopicSubscriber. May be the calling class itself, use reference "this".
     * @param topic A topic you wish to subscribe to, e.g "news", "myTopic". Topics are hierarchical, and can be specified on the form "news.politics.international". If subscribing to "news.politics", both "news.politics.international" as well as "news.politics.domestic" will be received.
     * The method has no return-value. All matching messages will be delivered to the TopicSubscriber specified, in an asynchronous manner (by a separate thread) 
     */
    public void subscribe(TopicSubscriber topicSubscriber, String topic) {

        SubscriptionHandler.getInstance().addSubscription(topic, topicSubscriber);

    }

     /**
     * Unsubscribes to a topic.
     * @param topicSubscriber Any instance of a class implementing TopicSubscriber. May be the calling class itself, use reference "this".
     * @param topic The topic you whish to unsubscribe to.
     */
    public void unsubscribe(TopicSubscriber topicSubscriber, String topic) {
        SubscriptionHandler.getInstance().removeSubscription(topic, topicSubscriber);
    }

     /**
     * Should be used when secure publish/subscribe communication is needed. Must be called before Invoker.subscribe().
     * @param password The string representing the password. Should match the password used when publisher calls Invoker.publishSecure().
     */
    public void setPublishSubscribePassword(String password) {
        SubscriptionHandler.getInstance().setPublishSubscribePassword(password);
    }


      /**
     * Publishes a message (an object) to the specified topic.
     * @param topic e.g "news", "myTopic". Topics are hierarchical, and can be specified on the form "news.politics.international". If subscribing to "news.politics", both "news.politics.international" as well as "news.politics.domestic" will be received.
     * @param object Any serializable object (may be a graph of objects) you wish to publish.
     */
     public void publish(String topic, Object object) {
        doPublish(topic, object, null);
     }


     /**
     * Publishes a message (an object) to the specified topic, using encrypted communication.
     * @param topic e.g "news", "myTopic". Topics are hierarchical, and can be specified on the form "news.politics.international". If subscribing to "news.politics", both "news.politics.international" as well as "news.politics.domestic" will be received.
     * @param object Any serializable object (may be a graph of objects) you wish to publish.
      * @param object Any serializable object (may be a graph of objects) you wish to publish.
     */
    public void publishSecure(String topic, Object object, String password) {

        doPublish(topic, object, password);
    }

    private void doPublish(String topic, Object object, String password) {
         Invoker invoker = new Invoker();

         try {

             if (password!=null)
                invoker.invokeSecure("xxDeliverPublishedMessage", password, topic, new SubscriptionMessage(object));
             else
                invoker.invoke("xxDeliverPublishedMessage", topic, new SubscriptionMessage(object));



         } catch (Exception e) {
             e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
         }
    }



     /**
     * Sends a service discovery request to all active Invokables in the network. For performance reasons, all invokables within the same JVM sends ONE collective response. All requests are aggregated in the calling Invoker, and returned as a list of services.
      * @return ArrayList containing all discovered services.
     */
    public ServiceDescription[] discoverServices() throws Exception {
        return doDiscoverServices(false, null);
    }


     /**
     * Sends a service discovery request using encrypted communication to all active Invokables in the network. For performance reasons, all invokables within the same JVM sends ONE collective response. All requests are aggregated in the calling Invoker, and returned as a list of services.
      * @return ArrayList containing all discovered services.
     */
    public ServiceDescription[] discoverServicesSecure(String password) throws Exception {
        return doDiscoverServices(true, password);
    }

    private ServiceDescription[] doDiscoverServices(boolean useEncryption, String password) throws Exception{

        InvocationMessage serviceRequestMessage = new InvocationMessage();

        serviceRequestMessage.setType(InvocationMessage.SERVICE_REQUEST);
        serviceRequestMessage.setSerialNumber(new Date().getTime());

        byte[] buffer = Base64.encodeObject(serviceRequestMessage, Base64.GZIP).getBytes();

        if (useEncryption) {
            buffer = UdpUtils.encryptBuffer(buffer, password);
        }


        UdpUtils.sendBuffer(buffer, transportType, unicastAddress, port);


        return receiveServiceResponse(serviceRequestMessage);

        
    }
}
